IoT CoreのトピックでJSON以外を受け取ってLambdaに渡してみた

IoT Coreにデバイスからデータが送られてくるとき、JSONフォーマットとは限りません。 IoT CoreのトピックはJSON以外のデータも受信できますが、ルールアクションでLambdaを実行したときのデータ受け取り方法が少し特殊でした。

具体的には、Lambdaが実行されなかったり、Lambdaは動くけど必ず失敗したりします。 そこで、JSON以外のデータでもLambdaが動く方法を試してみました。

実験準備

IoT Coreのルールで使用したSQL

IoT Coreのルールで次のSQLを使用し、Lambdaにデータを渡します。

SELECT * FROM '/test/hoge'

確認用のLambda

次のLambdaを作って、渡されたデータをログで確認します。

def lambda_handler(event, context):
    print(event)

Lambdaが実行すらされない事例

Publishしたデータ

下記のデータを/test/hogeトピックにPublishします。

"this is a pen."

結果

Lambdaは実行すらされませんでした。 CloudWatch Logsのルールアクションのメトリクスを見ると、実行に失敗していました。

Lambdaは実行されるが失敗する事例

Publishしたデータ

下記のデータを/test/hogeトピックにPublishします。

{"id": 1111, "name": "yamada"}{"id": 2222, "name": "yamada"}{"id": 3333, "name": "yamada"}

結果

Lambdaは実行されますが、下記ログが出力されており問答無用でLambda実行が失敗します。 （リトライ処理が働くため、Lambdaは計3回実行されます）

[ERROR] Runtime.UnmarshalError: Unable to unmarshal input: Extra data: line 1 column 31 (char 30)

対応策（LambdaにJSON以外のデータを渡す）

IoT Coreで受信した生データ（JSON以外）をそのままLambdaに渡しているのがNGであるため、Base64エンコードしてから渡します。 そのために下記のSQLを使います。

SELECT encode(*,'base64') as payload FROM '/test/hoge'

試してみる（その1）

Publishしたデータ

下記のデータを/test/hogeトピックにPublishします。

"this is a pen."

結果

Lambdaで下記のデータを受け取れました。

{'payload': 'dGhpcyBpcyBhIHBlbi4='}

これはBase64エンコードされているため、Base64デコードすればPublishされたデータと一致します。

>>> import base64
>>>
>>> base64.b64decode('dGhpcyBpcyBhIHBlbi4=').decode()
'this is a pen.'

試してみる（その2）

Publishしたデータ

下記のデータを/test/hogeトピックにPublishします。

{"id": 1111, "name": "yamada"}{"id": 2222, "name": "yamada"}{"id": 3333, "name": "yamada"}

結果

Lambdaで下記のデータを受け取れました。

{'payload': 'eyJpZCI6IDExMTEsICJuYW1lIjogInlhbWFkYSJ9eyJpZCI6IDIyMjIsICJuYW1lIjogInlhbWFkYSJ9eyJpZCI6IDMzMzMsICJuYW1lIjogInlhbWFkYSJ9'}

こちらもPublishされたデータと一致しました。

>>> import base64
>>>
>>> base64.b64decode('eyJpZCI6IDExMTEsICJuYW1lIjogInlhbWFkYSJ9eyJpZCI6IDIyMjIsICJuYW1lIjogInlhbWFkYSJ9eyJpZCI6IDMzMzMsICJuYW1lIjogInlhbWFkYSJ9').decode()
'{"id": 1111, "name": "yamada"}{"id": 2222, "name": "yamada"}{"id": 3333, "name": "yamada"}'

さいごに

地味に詰まる部分だと思うので、何らかの参考になれば幸いです。

参考

• SELECT 句 - AWS IoT
• encode()関数 - AWS IoT